package cn.tedu.test.charging.rabbitmq;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeoutException;

/**
 * 实现消费逻辑的测试类
 */
public class Consumer {
    //管理一个连接对象
    private Channel channel;
    @Before
    public void init() throws IOException, TimeoutException {
        //创建一个连接工厂,提供工厂的参数 连接rabbtimq的参数
        ConnectionFactory factory = new ConnectionFactory();
        //rabbitmq-server服务端 ip 与 端口 TCP协议
        factory.setHost("192.168.8.100");
        factory.setPort(5672);
        //提供用户名密码 本地local guest/guest 虚拟机admin/123456
        factory.setUsername("admin");
        factory.setPassword("123456");
        //rabbitmq virtual做数据隔离 连接的虚拟主机不一样,能够使用的交换机 队列 消息 是不同
        factory.setVirtualHost("/");
        Connection connection = factory.newConnection();
        channel= connection.createChannel();
    }
    //监听绑定 queue1
    @Test
    public void consume01() throws IOException {
        //给消费方法创建2个匿名内部类对象
        DeliverCallback deliver=new DeliverCallback() {
            /**
             * 消息消费逻辑方法
             * @param consumerTag 消费者唯一标识
             * @param message  传达过来 推送过来的队列的消息
             * @throws IOException 主要连接 读写异常
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                System.out.println("消费者1tag:"+consumerTag);
                //判断消费资源 是否达到阈值 临界值
                //解析消息 Delivery 包装了传递过来的消息数据 包括 消息内容body 包括 消息属性 properties 消息来源 路由
                byte[] body = message.getBody();//反序列化的消息信息
                message.getProperties();//生产者生成发送消息携带的属性
                message.getEnvelope().getExchange();
                message.getEnvelope().getRoutingKey();
                //将字符串解析
                String msg=new String(body);
                System.out.println("消费者1收到消息:"+msg+",消息消费处理时间:"+new Date().toLocaleString());
                //手动确认,当前消费者 消费的这条消息,是成功还是失败
                //消息确认 处理delivery 消息唯一标识符 告知消息队列哪个消息 被确认了
                //参数1: deliveryTag 唯一标识符
                //参数2: multiple 是否批量确认,如果true,一次性 批量确认,一个消费逻辑 处理多条消息,
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            }
        };
        CancelCallback cancel=new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };
        //调用短链接的消费方法 底层消费代码的连接 不能同时监听2个以上的消息队列 智能监听一个
        /**
         * String queue: 队列名称,当前连接 对象监听消费的队列是谁
         * Boolean autoAck:
         * 是否自动确认消息,如果true,消费者只要拿到消息,消息就会被删除,false 队列只有等到消费者的反馈确认 才会删除
         * DeliverCallback deliverCallback: 消费者拿到消息的回调方法.在这个方法中,是消息消费逻辑
         * CancelCallback cancelCallback:
         * 消费者取消消费的回调方法.消费者由于某种原因 链接队列拿到消息还没消费,取消监听消费.
         */
        channel.basicConsume("DL_QUEUE",false,deliver,cancel);
        //如果直接运行消费方法,测试框架执行方法结束,测试程序也就停止了,很难观察到一些运行逻辑
        while(true);
    }










}
